package cn.edu.hitwh.kanlema.service.impl;

import cn.edu.hitwh.kanlema.constant.MqConstants;
import cn.edu.hitwh.kanlema.constant.RedisConstants;
import cn.edu.hitwh.kanlema.entity.Cinema;
import cn.edu.hitwh.kanlema.exception.ApiException;
import cn.edu.hitwh.kanlema.mapper.CinemaMapper;
import cn.edu.hitwh.kanlema.message.RebuildCacheMessage;
import cn.edu.hitwh.kanlema.service.ICinemaService;
import cn.edu.hitwh.kanlema.util.RedisCache;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author vvcoder
 * @since 2025-8-18
 */
@Service
public class CinemaServiceImpl extends ServiceImpl<CinemaMapper, Cinema> implements ICinemaService {

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private RedissonClient redissonClient;

    private RBloomFilter<Long> cinemaBloomFilter;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;



    @PostConstruct
    public void initCinemaBloomFilter() {
        cinemaBloomFilter = redissonClient.getBloomFilter(RedisConstants.CACHE_CINEMA_BLOOM_KEY);
        long total = baseMapper.selectCount(null);
        RedisCache.initBloomFilter(
                cinemaBloomFilter,
                RedisConstants.CACHE_CINEMA_BLOOM_EXPECTED_INSERTIONS,
                RedisConstants.CACHE_CINEMA_BLOOM_FALSE_PROBABILITY,
                new RedisCache.DataLoader<Long>() {
                    long added = 0;

                    @Override
                    public List<Long> load(Long start, Long count) {
                        if (added >= total) {
                            return Collections.emptyList();
                        }
                        List<Long> ids = lambdaQuery()
                                .select(Cinema::getId)
                                .page(new Page<>(start / count, count))
                                .getRecords()
                                .stream()
                                .map(Cinema::getId)
                                .collect(Collectors.toList());
                        added += ids.size();
                        return ids;
                    }
                }
        );
    }

    @Override
    public boolean mightExist(Long id) {
        return cinemaBloomFilter.contains(id);
    }

    @Override
    public Cinema getCinemaById(Long id) {
        // 先检查布隆过滤器是否包含该id，若不包含则直接抛出异常
        if (!cinemaBloomFilter.contains(id)) {
            throw new ApiException("影院不存在");
        }
        Cinema cinema;
        if (isCinemaHot(id)) {
            // 热门影院，走逻辑过期缓存
            cinema = redisCache.getWithLogicalExpire(
                    RedisConstants.CACHE_HOT_CINEMA_KEY + id,
                    Cinema.class,
                    () -> {
                        String lockKey = RedisConstants.HOT_KEY_LOCK_PREFIX + RedisConstants.CACHE_HOT_CINEMA_KEY + id;
                        if (redisCache.tryLock(lockKey, RedisConstants.HOT_KEY_LOCK_TTL, TimeUnit.SECONDS)) {
                            RebuildCacheMessage rebuildMessage = new RebuildCacheMessage();
                            rebuildMessage.setCacheKey(RedisConstants.CACHE_HOT_CINEMA_KEY + id);
                            rebuildMessage.setLockKey(lockKey);
                            // 发送消息到队列
                            rabbitTemplate.convertAndSend(
                                    MqConstants.HOT_KEY_REBUILD_EXCHANGE,
                                    MqConstants.HOT_CINEMA_REBUILD_KEY,
                                    rebuildMessage
                            );
                        }
                    }
            );
        } else {
            // 普通影院，走旁路缓存
            cinema = redisCache.getObjectOrSet(
                    RedisConstants.CACHE_CINEMA_KEY + id,
                    () -> baseMapper.selectById(id),
                    Cinema.class,
                    RedisConstants.CACHE_CINEMA_TTL,
                    TimeUnit.MINUTES,
                    RedisConstants.CACHE_NULL_TTL // 缓存空值
            );
        }
        if (cinema == null) {
            throw new ApiException("影院不存在");
        }
        return cinema;
    }

    @Override
    public void saveCinema(Cinema cinema) {
        // 写入数据库
        baseMapper.insert(cinema);
        // 写入布隆过滤器
        cinemaBloomFilter.add(cinema.getId());
        // 清空缓存
        redisCache.delete(RedisConstants.CACHE_CINEMA_KEY + cinema.getId());
    }

    @Override
    public void updateCinemaById(Cinema cinema) {
        // 写入数据库
        baseMapper.updateById(cinema);
        // 清空缓存
        redisCache.delete(RedisConstants.CACHE_CINEMA_KEY + cinema.getId());
    }

    @Override
    public boolean isCinemaHot(Long id) {
        return Boolean.TRUE.equals(redisTemplate.opsForSet()
                .isMember(RedisConstants.CACHE_HOT_CINEMA_SET_KEY, id.toString()));
    }

    @Override
    public void setCinemaHot(Long id, Boolean isHot) {
        if (Boolean.TRUE.equals(isHot)) {
            Cinema cinema = getCinemaById(id);
            // 写入缓存
            redisCache.setWithLogicalExpire(
                    RedisConstants.CACHE_HOT_CINEMA_KEY + id,
                    cinema,
                    RedisConstants.CACHE_HOT_CINEMA_TTL,
                    TimeUnit.MINUTES
            );
            redisTemplate.opsForSet()
                    .add(RedisConstants.CACHE_HOT_CINEMA_SET_KEY, id.toString());
        } else {
            redisTemplate.opsForSet()
                    .remove(RedisConstants.CACHE_HOT_CINEMA_SET_KEY, id.toString());
            // 清空缓存
            redisCache.delete(RedisConstants.CACHE_HOT_CINEMA_KEY + id);
        }
    }
}
